package cn.dmp.report

import java.util.Properties

import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/4/22.
  */
object TerritoryAnalyze {

  def main(args: Array[String]): Unit = {

    //判断路径参数
    if (args.length!=1){
      println(
        """
          |inputDataPath
          |参数路径不合法
        """.stripMargin)
      sys.exit()
    }
    //创建sparkContext
    val conf = new SparkConf()
      .setMaster("local[*]").setAppName("TerritoryAnalyze")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)

    //输入 输出的路径参数
    val Array(inputDataPath)=args

    //读parquent文件要用SQLContext
    val sqlContext: SQLContext = new SQLContext(sc)
    val dataFrame = sqlContext.read.parquet(inputDataPath)

    dataFrame.registerTempTable("parquet")

    val sql: DataFrame = sqlContext.sql(
      """
      select provincename,cityname,
      sum(case when requestmode=1 and processnode>=1 then 1 else 0 end) 原始请求,
      sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) 有效请求,
      sum(case when requestmode=1 and processnode =3 then 1 else 0 end) 广告请求,
      sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid !=0 then 1 else 0 end) 参与竞价,
      sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) 竞价成功,
      sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) 展示数,
      sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) 点击数,
      sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) DSP广告消费,
      sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) DSP广告成本
      from parquet
      group by provincename,cityname
      """.stripMargin)

    //写进 mysql数据库
    val url="jdbc:mysql://192.168.20.13:3306/spark?characterEncoding=utf-8"
    val tname="TerritoryAnalyze"
    val prop = new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","06141ybw")
    prop.setProperty("driver","com.mysql.jdbc.Driver")
    sql.write.mode(SaveMode.Overwrite).jdbc(url,tname,prop)

    sc.stop()
  }

}
